package com.swhy

import java.net.URI
import java.util.{Date, UUID}

import com.utils._
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import org.codehaus.jackson.map.ObjectMapper


object ChgeAcctLblReason2 {
  def main(args: Array[String]): Unit = {
    if (args.length != 3) {
      println(
        """
          |com.swhy.CompanyInfo
          |params:
          | InputDate    刷数日期
          | OutputPath   输出路径
          | SplitNum     文件生成数量
        """.stripMargin)
      sys.exit()
    }
    val conf = new Configuration()
    val Array(inputDt, outputPath , splitNum) = args
    val sparkConf = new SparkConf().setAppName(this.getClass.getName)
    val sc = new SparkContext(sparkConf)
    val hiveContext = new HiveContext(sc)
//改下方的sql为需要装换成jason的sql
    val allcompany =s"""
        |select
        |ecif_cd,
        |cust_no,
        |id as capi_acc,
        |mgr_cd,
        |lbl_busi_type,
        |cast(lbl_type as string) as lbl_type,
        |lbl_name,
        |lbl_type_no,
        |lbl_val,
        |if(indx_resn is null,'[]', indx_resn) as ind_pub_reasn,
        |null as ind_reasn,
        |tm,
        |tm_flag_type,
        |cast(biz_date as string) as tm_flag
        |from
        |sum_data.sum_dao_lbl_custchg_d_q
        |where biz_date=${inputDt}
        |and id_type= 'a0101'
      """.stripMargin
    print(allcompany)
    val frame = hiveContext.sql(allcompany)
    val result = frame.rdd.map(x => {
      var cominfo = new StringBuilder()

      val ecif_cd = x.getAs[String]("ecif_cd")
      val cust_no = x.getAs[String]("cust_no")
      val capi_acc = x.getAs[String]("capi_acc")
      val mgr_cd = x.getAs[String]("mgr_cd")
      val lbl_busi_type = x.getAs[String]("lbl_busi_type")
      val lbl_name = x.getAs[String]("lbl_name")
      val lbl_type_no = x.getAs[String]("lbl_type_no")
      val lbl_val = x.getAs[String]("lbl_val")
      val lbl_type = x.getAs[String]("lbl_type")
      val ind_pub_reasn = x.getAs[String]("ind_pub_reasn")
      val ind_reasn = x.getAs[String]("ind_reasn")
      val tm = x.getAs[Date]("tm")
      val tm_flag_type = x.getAs[String]("tm_flag_type")


      //val _id = new ID()
      //val index = new Index()
      val companyInfo = new ChgeAcctLblReason()

      //_id.set_id(UUID.randomUUID().toString)
      //index.setIndex(_id)
      companyInfo.setEcifCd(ecif_cd)
      companyInfo.setCustNo(cust_no)
      companyInfo.setCapiAcc(capi_acc)
      companyInfo.setMgrCd(mgr_cd)
      companyInfo.setLblBusiType(lbl_busi_type)
      companyInfo.setLblType(lbl_type)
      companyInfo.setLblName(lbl_name)
      companyInfo.setLblTypeNo(lbl_type_no)
      companyInfo.setLblVal(lbl_val)
      companyInfo.setIndPubReasn(ind_pub_reasn)
      companyInfo.setIndReasn(ind_reasn)
      companyInfo.setTm(tm)
      companyInfo.setTmFlagType(tm_flag_type)

      val mapper = new ObjectMapper()

      //val indexJson = mapper.writeValueAsString() //企业自增编码json串
      val companyInfoJson = mapper.writeValueAsString(companyInfo) //企业信息串

      cominfo = cominfo.append(companyInfoJson) //json结果串
      cominfo.toString()
    })
    val path = new Path(outputPath) //创建文本输出路径
    val fs = FileSystem.get(URI.create(outputPath), conf)
    if (fs.exists(path)) {
      fs.delete(path, true)
    }
    result.repartition(splitNum.toInt).saveAsTextFile(outputPath)
    sc.stop()
  }
}
